package com.celeritymq.session;

import java.net.SocketAddress;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import javax.jms.JMSException;
import javax.jms.Queue;
import javax.jms.QueueBrowser;
import javax.jms.QueueReceiver;
import javax.jms.QueueSender;
import javax.jms.QueueSession;
import javax.jms.TemporaryQueue;

import com.celeritymq.browsers.CelerityQueueBrowser;
import com.celeritymq.consumer.CelerityQueueReceiver;
import com.celeritymq.destination.CelerityQueue;
import com.celeritymq.destination.CelerityTemporaryQueue;
import com.celeritymq.message.CelerityMessage;
import com.celeritymq.producer.CelerityQueueSender;
import com.celeritymq.utils.CelerityDestinationNameGenerator;

public class CelerityQueueSession extends CeleritySession implements QueueSession {
	private Map<String, CelerityQueueSender> senderMap = new HashMap<String, CelerityQueueSender>();
	private Map<String, CelerityQueueReceiver> receiverMap = new HashMap<String, CelerityQueueReceiver>();
	private Map<String, CelerityQueueBrowser> browserMap = new HashMap<String, CelerityQueueBrowser>();
	private Map<String, CelerityQueue> queueMap = new HashMap<String, CelerityQueue>();
	private Map<String, CelerityTemporaryQueue> temporaryQueueMap = new HashMap<String, CelerityTemporaryQueue>();
	private static final String NAME = "queue";
	private Map<String, BlockingQueue<CelerityMessage>> queueOut = new HashMap<String, BlockingQueue<CelerityMessage>>();

	public CelerityQueueSession(SocketAddress clientAddress, boolean transacted, int acknowledgeMode) {
		super(clientAddress);
		this.transacted = transacted;
		this.acknowledgeMode = acknowledgeMode;
	}

	@Override
	public QueueBrowser createBrowser(Queue queue) throws JMSException {
		return createBrowser(queue, null);
	}

	@Override
	public QueueBrowser createBrowser(Queue queue, String messageSelector) throws JMSException {
		String queueName = queue.getQueueName();
		CelerityQueueBrowser qb = browserMap.get(queueName);
		if (qb == null) {
			qb = new CelerityQueueBrowser(queue, messageSelector);
			browserMap.put(queueName, qb);
		}
		return qb;
	}

	@Override
	public Queue createQueue(String queueName) throws JMSException {
		CelerityQueue cq = queueMap.get(queueName);
		if (cq == null) {
			cq = new CelerityQueue(queueName);
			queueMap.put(queueName, cq);
		}
		return cq;
	}

	@Override
	public QueueReceiver createReceiver(Queue queue) throws JMSException {
		return createReceiver(queue, null);
	}

	@Override
	public QueueReceiver createReceiver(Queue queue, String messageSelector) throws JMSException {
		if (queue instanceof CelerityQueue) {
			throw new JMSException("not celerityqueue");
		}
		String queueName = queue.getQueueName();
		CelerityQueueReceiver cqr = new CelerityQueueReceiver((CelerityQueue) queue, messageSelector);
		receiverMap.put(queueName, cqr);
		return cqr;
	}

	@Override
	public QueueSender createSender(Queue queue) throws JMSException {
		String queueName = queue.getQueueName();
		CelerityQueueSender cqs = new CelerityQueueSender(queue);
		senderMap.put(queueName, cqs);
		return cqs;
	}

	public void putMessage(CelerityQueue queue, CelerityMessage message) throws JMSException {
		String queueName = queue.getQueueName();
		BlockingQueue<CelerityMessage> messageQueue = queueOut.get(queueName);
		if (messageQueue == null) {
			messageQueue = new LinkedBlockingQueue<CelerityMessage>();
			queueOut.put(queueName, messageQueue);
		}
		try {
			messageQueue.put(message);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	@Override
	public TemporaryQueue createTemporaryQueue() throws JMSException {
		String tempQueueName = null;
		do {
			tempQueueName = CelerityDestinationNameGenerator.generateName(NAME);
		} while (tempQueueName == null || !temporaryQueueMap.containsKey(tempQueueName));
		CelerityTemporaryQueue ctq = new CelerityTemporaryQueue(this, tempQueueName);
		temporaryQueueMap.put(tempQueueName, ctq);
		return ctq;
	}

}
